
package cn.tang.tframe.mq.consumer;

import cn.tang.tframe.common.base.MainConfigLoader;
import cn.tang.tframe.common.utils.io.PropertiesUtil;
import cn.tang.tframe.common.utils.lang.SpiUtils;
import cn.tang.tframe.common.utils.lang.StringUtils;
import cn.tang.tframe.dynconfig.DynConfigRegister;
import cn.tang.tframe.dynconfig.IDynConfigProcessor;
import cn.tang.tframe.dynconfig.parser.ConfigValueParser;
import cn.tang.tframe.dynconfig.parser.StringConfigValueParser;
import cn.tang.tframe.mq.IQtMqFactory;
import cn.tang.tframe.mq.kafka.producer.KafkaProducerClient;
import cn.tang.tframe.mq.producer.IQtProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * Created by tanghc on 17/1/20.
 */
public class QtConsumerClient {

	static Logger logger = LoggerFactory.getLogger(KafkaProducerClient.class);

	private QtConsumerClient() {
	}

	private static class QtProducerClientHolder {

		private static IQtProducer instance;
		private static final String mq_config_group = "common";
		private static final String mq_config_dataId = "mqConfig";
		private static String mq = "kafka";
		private static boolean isInit = false;
		private static final String productCode =
			MainConfigLoader.getInstance().getProductCode();
		private static final String appCode =
			MainConfigLoader.getInstance().getAppCode();
		static {
			DynConfigRegister.getInstance().buildOnce(
				productCode, appCode, mq_config_group, mq_config_dataId,
				new IDynConfigProcessor() {

					@Override
					public void procesConfig(String config) {

						Properties properties =
							PropertiesUtil.loadFromString(config);
						ConfigValueParser<String> configValueParser =
							new StringConfigValueParser();
						mq = configValueParser.parse(
							properties, "consumer", "kafka");
					}
				});
		}

		private static Runnable builderConsumer(
			String rawTopic, Properties properties,
			IQtMsgProcessor msgProcessor) {

			IQtMqFactory mqFactory =
				SpiUtils.getSpiBean(mq, IQtMqFactory.class);
			if (null != mqFactory) {
				return mqFactory.builderConsumer(
					rawTopic, properties, msgProcessor);
			}
			else {
				throw new RuntimeException("no qtmqfactory spi:" + mq);
			}
		}

	}

	public static Runnable builderConsumer(
		String rawTopic, IQtMsgProcessor msgProcessor) {

		return builderConsumer(rawTopic, null, msgProcessor);
	}

	public static Runnable builderConsumer(
		String projectCode, String appCode, String topic,
		IQtMsgProcessor msgProcessor) {

		return builderConsumer(projectCode,appCode,topic,null,msgProcessor);
	}

	public static Runnable builderConsumer(
		String rawTopic, Properties properties, IQtMsgProcessor msgProcessor){
		return QtProducerClientHolder.builderConsumer(rawTopic,properties,msgProcessor);
	}

	public static Runnable builderConsumer(
		String projectCode, String appCode, String topic, Properties properties,
		IQtMsgProcessor msgProcessor){
		return builderConsumer(StringUtils.join(new Object[] {
				projectCode, appCode, topic
		}, '_'), properties, msgProcessor);
	}
}
